// Copyright(C) 2024 InfiniFlow, Inc. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

module;

#include <cassert>

module infinity_core:http_search.impl;

import :http_search;
import :infinity;
import :data_table;
import :defer_op;
import :query_result;
import :data_block;
import :value;
import :physical_import;
import :utility;

import std;

import parsed_expr;
import knn_expr;
import match_expr;
import search_expr;
import match_tensor_expr;
import match_sparse_expr;
import fusion_expr;
import column_expr;
import function_expr;
import constant_expr;
import expr_parser;
import expression_parser_result;
import statement_common;
import explain_statement;
import internal_types;
import select_statement;
import logical_type;

namespace infinity {

void HTTPSearch::Process(Infinity *infinity_ptr,
                         const std::string &db_name,
                         const std::string &table_name,
                         const std::string &input_json_str,
                         HTTPStatus &http_status,
                         nlohmann::json &response) {
    http_status = HTTPStatus::CODE_500;
    try {
        simdjson::padded_string json_pad(input_json_str);
        simdjson::parser parser;
        simdjson::document doc = parser.iterate(json_pad);
        if (doc.type() != simdjson::json_type::object) {
            response["error_code"] = ErrorCode::kInvalidJsonFormat;
            response["error_msg"] = "HTTP Body isn't json object";
        }
        std::unique_ptr<ParsedExpr> filter{};
        std::unique_ptr<ParsedExpr> limit{};
        std::unique_ptr<ParsedExpr> offset{};
        SearchExpr *search_expr{};
        std::vector<ParsedExpr *> *output_columns{nullptr};
        std::vector<ParsedExpr *> *highlight_columns{nullptr};
        std::vector<OrderByExpr *> *order_by_list{nullptr};
        std::vector<ParsedExpr *> *group_by_columns{nullptr};
        std::unique_ptr<ParsedExpr> having{};
        bool total_hits_count_flag{};

        DeferFn search_fn([&]() {
            if (search_expr != nullptr) {
                delete search_expr;
                search_expr = nullptr;
            }
        });

        DeferFn defer_fn([&]() {
            if (output_columns != nullptr) {
                for (auto &expr : *output_columns) {
                    delete expr;
                }
                delete output_columns;
                output_columns = nullptr;
            }
        });

        DeferFn defer_fn_highlight([&]() {
            if (highlight_columns != nullptr) {
                for (auto &expr : *highlight_columns) {
                    delete expr;
                }
                delete highlight_columns;
                highlight_columns = nullptr;
            }
        });

        DeferFn defer_fn_order([&]() {
            if (order_by_list != nullptr) {
                for (auto &expr : *order_by_list) {
                    delete expr;
                }
                delete order_by_list;
                order_by_list = nullptr;
            }
        });

        for (auto elem : doc.get_object()) {
            auto key = std::string(static_cast<std::string_view>(elem.unescaped_key()));
            auto value = elem.value();
            ToLower(key);
            if (key == "output") {
                if (output_columns != nullptr) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one output field.";
                    return;
                }
                if (value.type() != simdjson::json_type::array) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "Output field should be array";
                    return;
                }

                output_columns = ParseOutput(value.raw_json(), http_status, response);
                if (output_columns == nullptr) {
                    return;
                }
            } else if (key == "highlight") {
                if (value.type() != simdjson::json_type::array) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "Output field should be array";
                    return;
                }

                highlight_columns = ParseOutput(value.raw_json(), http_status, response);
                if (highlight_columns == nullptr) {
                    return;
                }
            } else if (key == "sort") {
                if (order_by_list != nullptr) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one sort field.";
                    return;
                }

                if (value.type() != simdjson::json_type::array) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "Sort field should be array";
                    return;
                }

                order_by_list = ParseSort(value.raw_json(), http_status, response);
                if (order_by_list == nullptr) {
                    return;
                }
            } else if (key == "group_by") {
                if (group_by_columns != nullptr) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one group by field.";
                    return;
                }

                group_by_columns = ParseOutput(value.raw_json(), http_status, response);
                if (group_by_columns == nullptr) {
                    return;
                }
            } else if (key == "having") {
                if (having != nullptr) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one having field.";
                    return;
                }
                having = ParseFilter(value.raw_json(), http_status, response);
                if (having == nullptr) {
                    return;
                }
            } else if (key == "filter") {

                if (filter) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one filter field.";
                    return;
                }
                filter = ParseFilter(value.raw_json(), http_status, response);
                if (!filter) {
                    return;
                }
            } else if (key == "limit") {

                if (limit) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one limit field.";
                    return;
                }
                limit = ParseFilter(value.raw_json(), http_status, response);
                if (!limit) {
                    return;
                }
            } else if (key == "offset") {

                if (offset) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one offset field.";
                    return;
                }
                offset = ParseFilter(value.raw_json(), http_status, response);
                if (!offset) {
                    return;
                }
            } else if (key == "search") {
                if (search_expr) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one search field.";
                    return;
                }
                search_expr = ParseSearchExpr(value.raw_json(), http_status, response);
                if (!search_expr) {
                    return;
                }
            } else if (key == "option") {
                if (value.type() != simdjson::json_type::object) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "Option field should be object";
                    return;
                }

                for (auto option : value.get_object()) {
                    auto option_key = std::string(static_cast<std::string_view>(option.unescaped_key()));
                    auto option_value = option.value();
                    ToLower(option_key);
                    if (option_key == "total_hits_count") {
                        if (option_value.is_string()) {
                            std::string value_str = option_value.get<std::string>();
                            [[maybe_unused]] auto maybe_a_bug = std::tolower(value);
                            if (value_str == "true") {
                                total_hits_count_flag = true;
                            } else if (value_str == "false") {
                                total_hits_count_flag = false;
                            } else {
                                response["error_code"] = ErrorCode::kInvalidExpression;
                                response["error_msg"] = fmt::format("Unknown search option: {}, value: {}", key, value_str);
                                return;
                            }
                        } else if (option_value.type() == simdjson::json_type::boolean) {
                            total_hits_count_flag = option_value.get<bool>();
                        } else {
                            response["error_code"] = ErrorCode::kInvalidExpression;
                            response["error_msg"] = "Invalid total hits count type";
                            return;
                        }
                    }
                }
            } else {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "Unknown expression: " + key;
                return;
            }
        }

        const QueryResult result = infinity_ptr->Search(db_name,
                                                        table_name,
                                                        search_expr,
                                                        filter.release(),
                                                        limit.release(),
                                                        offset.release(),
                                                        output_columns,
                                                        highlight_columns,
                                                        order_by_list,
                                                        group_by_columns,
                                                        having.release(),
                                                        total_hits_count_flag);

        search_expr = nullptr;
        output_columns = nullptr;
        highlight_columns = nullptr;
        order_by_list = nullptr;
        group_by_columns = nullptr;
        if (result.IsOk()) {
            size_t block_rows = result.result_table_->DataBlockCount();
            for (size_t block_id = 0; block_id < block_rows; ++block_id) {
                DataBlock *data_block = result.result_table_->GetDataBlockById(block_id).get();
                auto row_count = data_block->row_count();
                auto column_cnt = result.result_table_->ColumnCount();

                for (int row = 0; row < row_count; ++row) {
                    nlohmann::json json_result_row;
                    for (size_t col = 0; col < column_cnt; ++col) {
                        nlohmann::json json_result_cell;
                        Value value = data_block->GetValue(col, row);
                        const std::string &column_name = result.result_table_->GetColumnNameById(col);
                        switch (value.type().type()) {
                            case LogicalType::kTinyInt:
                            case LogicalType::kSmallInt:
                            case LogicalType::kInteger:
                            case LogicalType::kBigInt: {
                                json_result_cell[column_name] = value.ToInteger();
                                break;
                            }
                            case LogicalType::kFloat: {
                                json_result_cell[column_name] = value.ToFloat();
                                break;
                            }
                            case LogicalType::kDouble: {
                                json_result_cell[column_name] = value.ToDouble();
                                break;
                            }
                            default: {
                                json_result_cell[column_name] = value.ToString();
                                break;
                            }
                        }
                        json_result_row.push_back(json_result_cell);
                    }
                    response["output"].push_back(json_result_row);
                }
            }

            if (result.result_table_->total_hits_count_flag_) {
                response["total_hits_count"] = result.result_table_->total_hits_count_;
            }

            response["error_code"] = 0;
            http_status = HTTPStatus::CODE_200;
        } else {
            response["error_code"] = result.ErrorCode();
            response["error_msg"] = result.ErrorMsg();
            http_status = HTTPStatus::CODE_500;
        }
    } catch (simdjson::simdjson_error &e) {
        response["error_code"] = ErrorCode::kInvalidJsonFormat;
        response["error_msg"] = e.what();
    }
    return;
}

void HTTPSearch::Explain(Infinity *infinity_ptr,
                         const std::string &db_name,
                         const std::string &table_name,
                         const std::string &input_json_str,
                         HTTPStatus &http_status,
                         nlohmann::json &response) {
    http_status = HTTPStatus::CODE_500;
    try {
        simdjson::padded_string json_pad(input_json_str);
        simdjson::parser parser;
        simdjson::document doc = parser.iterate(json_pad);
        if (doc.type() != simdjson::json_type::object) {
            response["error_code"] = ErrorCode::kInvalidJsonFormat;
            response["error_msg"] = "HTTP Body isn't json object";
        }
        std::unique_ptr<ParsedExpr> filter{};
        std::unique_ptr<ParsedExpr> limit{};
        std::unique_ptr<ParsedExpr> offset{};
        SearchExpr *search_expr{};
        std::vector<ParsedExpr *> *output_columns{nullptr};
        std::vector<ParsedExpr *> *highlight_columns{nullptr};
        std::vector<OrderByExpr *> *order_by_list{nullptr};
        std::vector<ParsedExpr *> *group_by_columns{nullptr};
        std::unique_ptr<ParsedExpr> having{nullptr};
        DeferFn defer_fn([&]() {
            if (output_columns != nullptr) {
                for (auto &expr : *output_columns) {
                    delete expr;
                }
                delete output_columns;
                output_columns = nullptr;
            }
        });
        DeferFn defer_fn_highlight([&]() {
            if (highlight_columns != nullptr) {
                for (auto &expr : *highlight_columns) {
                    delete expr;
                }
                delete highlight_columns;
                highlight_columns = nullptr;
            }
        });

        DeferFn defer_fn_order([&]() {
            if (order_by_list != nullptr) {
                for (auto &expr : *order_by_list) {
                    delete expr;
                }
                delete order_by_list;
                order_by_list = nullptr;
            }
        });

        ExplainType explain_type = ExplainType::kInvalid;
        for (auto elem : doc.get_object()) {
            auto key = std::string(static_cast<std::string_view>(elem.unescaped_key()));
            auto value = elem.value();
            ToLower(key);
            if (key == "output") {
                if (output_columns != nullptr) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one output field.";
                    return;
                }
                if (value.type() != simdjson::json_type::array) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "Output field should be array";
                    return;
                }

                output_columns = ParseOutput(value.raw_json(), http_status, response);
                if (output_columns == nullptr) {
                    return;
                }
            } else if (key == "highlight") {
                if (value.type() != simdjson::json_type::array) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "Output field should be array";
                    return;
                }

                highlight_columns = ParseOutput(value.raw_json(), http_status, response);
                if (highlight_columns == nullptr) {
                    return;
                }
            } else if (key == "sort") {
                if (order_by_list != nullptr) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one sort field.";
                    return;
                }

                if (value.type() != simdjson::json_type::array) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "Sort field should be array";
                    return;
                }

                order_by_list = ParseSort(value.raw_json(), http_status, response);
                if (order_by_list == nullptr) {
                    return;
                }
            } else if (key == "group_by") {
                if (group_by_columns != nullptr) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one group by field.";
                    return;
                }

                group_by_columns = ParseOutput(value.raw_json(), http_status, response);
                if (group_by_columns == nullptr) {
                    return;
                }
            } else if (key == "having") {
                if (having != nullptr) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one having field.";
                    return;
                }
                having = ParseFilter(value.raw_json(), http_status, response);
                if (having == nullptr) {
                    return;
                }
            } else if (key == "filter") {

                if (filter != nullptr) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one output field.";
                    return;
                }
                filter = ParseFilter(value.raw_json(), http_status, response);
                if (filter == nullptr) {
                    return;
                }
            } else if (key == "limit") {

                if (limit) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one limit field.";
                    return;
                }
                limit = ParseFilter(value.raw_json(), http_status, response);
                if (!limit) {
                    return;
                }
            } else if (key == "offset") {

                if (offset) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one offset field.";
                    return;
                }
                offset = ParseFilter(value.raw_json(), http_status, response);
                if (!offset) {
                    return;
                }
            } else if (key == "search") {
                if (search_expr) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = "More than one search field.";
                    return;
                }
                search_expr = ParseSearchExpr(value.raw_json(), http_status, response);
                if (!search_expr) {
                    return;
                }
            } else if (key == "explain_type") {
                if (std::string type = value.get<std::string>(); type == "analyze") {
                    explain_type = ExplainType::kAnalyze;
                } else if (type == "ast") {
                    explain_type = ExplainType::kAst;
                } else if (type == "physical") {
                    explain_type = ExplainType::kPhysical;
                } else if (type == "pipeline") {
                    explain_type = ExplainType::kPipeline;
                } else if (type == "unopt") {
                    explain_type = ExplainType::kUnOpt;
                } else if (type == "opt") {
                    explain_type = ExplainType::kOpt;
                } else if (type == "fragment") {
                    explain_type = ExplainType::kFragment;
                } else {
                    explain_type = ExplainType::kInvalid;
                }
            } else {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "Unknown expression: " + key;
                return;
            }
        }

        const QueryResult result = infinity_ptr->Explain(db_name,
                                                         table_name,
                                                         explain_type,
                                                         search_expr,
                                                         filter.release(),
                                                         limit.release(),
                                                         offset.release(),
                                                         output_columns,
                                                         highlight_columns,
                                                         order_by_list,
                                                         group_by_columns,
                                                         having.release());
        search_expr = nullptr;
        output_columns = nullptr;
        highlight_columns = nullptr;
        order_by_list = nullptr;
        group_by_columns = nullptr;
        if (result.IsOk()) {
            size_t block_rows = result.result_table_->DataBlockCount();
            for (size_t block_id = 0; block_id < block_rows; ++block_id) {
                DataBlock *data_block = result.result_table_->GetDataBlockById(block_id).get();
                auto row_count = data_block->row_count();
                auto column_cnt = result.result_table_->ColumnCount();

                for (int row = 0; row < row_count; ++row) {
                    nlohmann::json json_result_row;
                    for (size_t col = 0; col < column_cnt; ++col) {
                        Value value = data_block->GetValue(col, row);
                        const std::string &column_name = result.result_table_->GetColumnNameById(col);
                        const std::string &column_value = value.ToString();
                        json_result_row[column_name] = column_value;
                    }
                    response["output"].push_back(json_result_row);
                }
            }

            response["error_code"] = 0;
            http_status = HTTPStatus::CODE_200;
        } else {
            response["error_code"] = result.ErrorCode();
            response["error_msg"] = result.ErrorMsg();
            http_status = HTTPStatus::CODE_500;
        }
    } catch (simdjson::simdjson_error &e) {
        response["error_code"] = ErrorCode::kInvalidJsonFormat;
        response["error_msg"] = e.what();
    }
    return;
}

std::unique_ptr<ParsedExpr> HTTPSearch::ParseFilter(std::string_view json_sv, HTTPStatus &http_status, nlohmann::json &response) {
    simdjson::padded_string json_pad(json_sv);
    simdjson::parser parser;
    simdjson::document doc = parser.iterate(json_pad);

    if (doc.type() != simdjson::json_type::string) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = "Filter field should be string";
        return nullptr;
    }
    std::unique_ptr<ExpressionParserResult> expr_parsed_result = std::make_unique<ExpressionParserResult>();
    ExprParser expr_parser;
    expr_parser.Parse(doc.get<std::string>(), expr_parsed_result.get());
    if (expr_parsed_result->IsError() || expr_parsed_result->exprs_ptr_->size() != 1) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = fmt::format("Invalid expression: {}", std::string(json_sv));
        return nullptr;
    }

    ParsedExpr *res = expr_parsed_result->exprs_ptr_->at(0);
    expr_parsed_result->exprs_ptr_->at(0) = nullptr;
    std::unique_ptr<ParsedExpr> filter_expr(res);
    return filter_expr;
}

std::vector<ParsedExpr *> *HTTPSearch::ParseOutput(std::string_view json_sv, HTTPStatus &http_status, nlohmann::json &response) {
    simdjson::padded_string json_pad(json_sv);
    simdjson::parser parser;
    simdjson::document doc = parser.iterate(json_pad);

    std::vector<ParsedExpr *> *output_columns = new std::vector<ParsedExpr *>();
    DeferFn free_output_columns([&]() {
        if (output_columns != nullptr) {
            for (auto &expr : *output_columns) {
                delete expr;
                expr = nullptr;
            }
            delete output_columns;
            output_columns = nullptr;
        }
    });

    std::vector<std::string> output_exprs;
    if (doc.type() == simdjson::json_type::array) {
        output_columns->reserve(doc.count_elements());
        for (auto output_expr : doc.get_array()) {
            if (!output_expr.is_string()) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = fmt::format("Invalid expression: {}", std::string(json_sv));
                return nullptr;
            }
            output_exprs.push_back(output_expr.get<std::string>());
        }
    } else if (doc.type() == simdjson::json_type::string) {
        output_columns->reserve(1);
        output_exprs.push_back(doc.get<std::string>());
    } else {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = fmt::format("Invalid expression: {}", std::string(json_sv));
        return nullptr;
    }

    for (auto output_expr : output_exprs) {
        std::string output_expr_str = output_expr;
        if (output_expr_str == "_row_id" or output_expr_str == "_similarity" or output_expr_str == "_distance" or output_expr_str == "_score") {
            auto parsed_expr = new FunctionExpr();
            if (output_expr_str == "_row_id") {
                parsed_expr->func_name_ = "row_id";
            } else if (output_expr_str == "_similarity") {
                parsed_expr->func_name_ = "similarity";
            } else if (output_expr_str == "_distance") {
                parsed_expr->func_name_ = "distance";
            } else if (output_expr_str == "_score") {
                parsed_expr->func_name_ = "score";
            }
            output_columns->emplace_back(parsed_expr);
            parsed_expr = nullptr;
        } else {
            std::unique_ptr<ExpressionParserResult> expr_parsed_result = std::make_unique<ExpressionParserResult>();
            ExprParser expr_parser;
            expr_parser.Parse(output_expr_str, expr_parsed_result.get());
            if (expr_parsed_result->IsError() || expr_parsed_result->exprs_ptr_->empty()) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = fmt::format("Invalid expression: {}", output_expr_str);
                return nullptr;
            }

            auto &expr_ptr = expr_parsed_result->exprs_ptr_->at(0);
            if (expr_ptr->type_ == ParsedExprType::kColumn) {
                if (output_expr_str == "*") {
                    auto *column_expr = static_cast<ColumnExpr *>(expr_ptr);
                    column_expr->star_ = true;
                }
            }

            output_columns->emplace_back(expr_ptr);
            expr_ptr = nullptr;
        }
    }

    // Avoiding DeferFN auto free the output expressions
    std::vector<ParsedExpr *> *res = output_columns;
    output_columns = nullptr;
    return res;
}

std::vector<OrderByExpr *> *HTTPSearch::ParseSort(std::string_view json_sv, HTTPStatus &http_status, nlohmann::json &response) {
    simdjson::padded_string json_pad(json_sv);
    simdjson::parser parser;
    simdjson::document doc = parser.iterate(json_pad);

    std::vector<OrderByExpr *> *order_by_list = new std::vector<OrderByExpr *>();
    DeferFn defer_fn([&]() {
        if (order_by_list != nullptr) {
            for (auto &expr : *order_by_list) {
                delete expr;
            }
            delete order_by_list;
            order_by_list = nullptr;
        }
    });

    for (auto order_expr : doc.get_array()) {
        for (auto expression : order_expr.get_object()) {
            std::string key = std::string((std::string_view)expression.unescaped_key());
            ToLower(key);
            auto order_by_expr = std::make_unique<OrderByExpr>();
            if (key == "_row_id" or key == "_similarity" or key == "_distance" or key == "_score") {
                auto parsed_expr = new FunctionExpr();
                if (key == "_row_id") {
                    parsed_expr->func_name_ = "row_id";
                } else if (key == "_similarity") {
                    parsed_expr->func_name_ = "similarity";
                } else if (key == "_distance") {
                    parsed_expr->func_name_ = "distance";
                } else if (key == "_score") {
                    parsed_expr->func_name_ = "score";
                }
                order_by_expr->expr_ = parsed_expr;
                parsed_expr = nullptr;
            } else {
                std::unique_ptr<ExpressionParserResult> expr_parsed_result = std::make_unique<ExpressionParserResult>();
                ExprParser expr_parser;
                expr_parser.Parse(key, expr_parsed_result.get());
                if (expr_parsed_result->IsError() || expr_parsed_result->exprs_ptr_->empty()) {
                    response["error_code"] = ErrorCode::kInvalidExpression;
                    response["error_msg"] = fmt::format("Invalid expression: {}", key);
                    return nullptr;
                }

                order_by_expr->expr_ = expr_parsed_result->exprs_ptr_->at(0);
                expr_parsed_result->exprs_ptr_->at(0) = nullptr;
            }

            std::string value = expression.value().get<std::string>();
            ToLower(value);
            if (value == "asc") {
                order_by_expr->type_ = OrderType::kAsc;
            } else if (value == "desc") {
                order_by_expr->type_ = OrderType::kDesc;
            } else {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = fmt::format("Invalid expression: {}", value);
                return nullptr;
            }

            order_by_list->emplace_back(order_by_expr.release());
        }
    }

    // Avoiding DeferFN auto free the output expressions
    std::vector<OrderByExpr *> *res = order_by_list;
    order_by_list = nullptr;
    return res;
}

SearchExpr *HTTPSearch::ParseSearchExpr(std::string_view json_sv, HTTPStatus &http_status, nlohmann::json &response) {
    simdjson::padded_string json_pad(json_sv);
    simdjson::parser parser;
    simdjson::document doc = parser.iterate(json_pad);

    if (doc.type() != simdjson::json_type::array) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = "Search field should be list";
        return {};
    }
    auto child_expr = new std::vector<ParsedExpr *>();
    DeferFn defer_fn([&] {
        if (child_expr) {
            for (const auto expr : *child_expr) {
                delete expr;
            }
            delete child_expr;
            child_expr = nullptr;
        }
    });
    child_expr->reserve(doc.count_elements());
    for (auto search_obj : doc.get_array()) {
        std::string_view search_obj_sv = search_obj.raw_json();
        simdjson::padded_string sub_pad(search_obj_sv);
        simdjson::parser sub_parser;
        simdjson::document sub_doc = sub_parser.iterate(sub_pad);

        if (sub_doc.type() != simdjson::json_type::object) {
            response["error_code"] = ErrorCode::kInvalidExpression;
            response["error_msg"] = "Search field should be list of objects";
            return {};
        }

        auto fusion = sub_doc["fusion_method"].get<std::string>();
        auto match = sub_doc["match_method"].get<std::string>();

        if (fusion.error() == simdjson::SUCCESS && match.error() == simdjson::SUCCESS) {
            response["error_code"] = ErrorCode::kInvalidExpression;
            response["error_msg"] = "Every single search expression should not contain both fusion_method and match_method fields";
            return {};
        }
        if (fusion.error() == simdjson::SUCCESS) {
            auto fusion_expr = ParseFusion(search_obj_sv, http_status, response);
            if (!fusion_expr) {
                return {};
            }
            child_expr->push_back(fusion_expr.release());
        } else {
            // match type
            std::string match_method = match.value();
            ToLower(match_method);
            if (match_method == "dense") {
                auto match_dense_expr = ParseMatchDense(search_obj_sv, http_status, response);
                if (!match_dense_expr) {
                    return {};
                }
                child_expr->push_back(match_dense_expr.release());
            } else if (match_method == "sparse") {
                auto match_sparse_expr = ParseMatchSparse(search_obj_sv, http_status, response);
                if (!match_sparse_expr) {
                    return {};
                }
                child_expr->push_back(match_sparse_expr.release());
            } else if (match_method == "text") {
                auto match_text_expr = ParseMatchText(search_obj_sv, http_status, response);
                if (!match_text_expr) {
                    return {};
                }
                child_expr->push_back(match_text_expr.release());
            } else if (match_method == "tensor") {
                auto match_tensor_expr = ParseMatchTensor(search_obj_sv, http_status, response);
                if (!match_tensor_expr) {
                    return {};
                }
                child_expr->push_back(match_tensor_expr.release());
            } else {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = fmt::format("Unknown match method: {}", match_method);
                return {};
            }
        }
    }
    auto search_expr = new SearchExpr();
    try {
        search_expr->SetExprs(child_expr);
        child_expr = nullptr;
    } catch (std::exception &e) {
        delete search_expr;
        search_expr = nullptr;
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = fmt::format("Invalid Search expression, error info: {}", e.what());
        return nullptr;
    }
    return search_expr;
}

std::unique_ptr<FusionExpr> HTTPSearch::ParseFusion(std::string_view json_sv, HTTPStatus &http_status, nlohmann::json &response) {
    simdjson::padded_string json_pad(json_sv);
    simdjson::parser parser;
    simdjson::document doc = parser.iterate(json_pad);

    if (doc.type() != simdjson::json_type::object) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = fmt::format("Fusion expression must be a json object: {}", std::string(json_sv));
        return nullptr;
    }
    i64 topn = -1;
    nlohmann::json fusion_params;
    auto fusion_expr = std::make_unique<FusionExpr>();
    // must have: "fusion_method", "topn"
    // may have: "params"
    static constexpr std::array<std::string, 3> possible_keys{"fusion_method", "topn", "params"};
    std::set<std::string> possible_keys_set(possible_keys.begin(), possible_keys.end());
    for (auto expression : doc.get_object()) {
        auto key = std::string(static_cast<std::string_view>(expression.unescaped_key()));
        auto params = expression.value();
        ToLower(key);
        if (!possible_keys_set.erase(key)) {
            response["error_code"] = ErrorCode::kInvalidExpression;
            response["error_msg"] = fmt::format("Unknown Fusion expression key: {}", key);
            return nullptr;
        }
        if (key == "fusion_method") {
            std::string method_str = params.get<std::string>();
            ToLower(method_str);
            fusion_expr->method_ = std::move(method_str);
        } else if (key == "topn") {
            if (!params.is_integer()) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "Fusion expression topn field should be integer";
                return nullptr;
            }
            topn = params.get<i64>();
        } else if (key == "params") {
            if (params.type() != simdjson::json_type::object) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "Fusion params should be object";
                return nullptr;
            }
            fusion_params = nlohmann::json::parse(static_cast<std::string_view>(params.raw_json()));
        }
    }
    // "params" is optional
    possible_keys_set.erase("params");
    // check if all required fields are set
    if (!possible_keys_set.empty()) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] =
            fmt::format("Invalid Fusion expression: following fields are required but not found: {}", fmt::join(possible_keys_set, ", "));
        return nullptr;
    }
    if (topn <= 0) {
        response["error_code"] = ErrorCode::kInvalidTopKType;
        response["error_msg"] = "Fusion expression topn field should be positive integer";
        return nullptr;
    }
    std::string extra_params_str{};
    if (fusion_expr->method_ == "match_tensor") {
        nlohmann::json match_tensor_json = fusion_params;
        match_tensor_json["match_method"] = "tensor";
        if (match_tensor_json.contains("topn")) {
            response["error_code"] = ErrorCode::kInvalidExpression;
            response["error_msg"] = "Fusion expression topn field should not be set in params";
            return nullptr;
        }
        match_tensor_json["topn"] = topn;
        auto match_tensor_expr = ParseMatchTensor(match_tensor_json.dump(), http_status, response);
        if (!match_tensor_expr) {
            return nullptr;
        }
        fusion_expr->match_tensor_expr_ = std::move(match_tensor_expr);
    } else {
        simdjson::padded_string fusion_pad(fusion_params.dump());
        simdjson::document fusion_doc = parser.iterate(fusion_pad);
        if (fusion_doc.type() == simdjson::json_type::object) {
            for (auto param : fusion_doc.get_object()) {
                std::string param_k = std::string((std::string_view)param.unescaped_key());
                std::string param_v = param.value().get<std::string>();
                extra_params_str += fmt::format(";{}={}", param_k, param_v);
            }
        }
    }
    fusion_expr->SetOptions(fmt::format("topn={}{}", topn, extra_params_str));
    return fusion_expr;
}

std::unique_ptr<KnnExpr> HTTPSearch::ParseMatchDense(std::string_view json_sv, HTTPStatus &http_status, nlohmann::json &response) {
    simdjson::padded_string json_pad(json_sv);
    simdjson::parser parser;
    simdjson::document doc = parser.iterate(json_pad);

    if (doc.type() != simdjson::json_type::object) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = "MatchDense field should be object";
        return nullptr;
    }
    auto knn_expr = std::make_unique<KnnExpr>();
    i64 topn = -1;
    std::string_view query_vector_json;
    std::string_view fde_json;
    bool has_query_vector = false;
    bool has_fde = false;
    // must have: "match_method", "fields", ("query_vector" OR "fde"), "element_type", "metric_type", "topn"
    // may have: "params"
    constexpr std::array possible_keys{"match_method", "fields", "query_vector", "fde", "element_type", "metric_type", "topn", "params"};
    std::set<std::string> possible_keys_set(possible_keys.begin(), possible_keys.end());
    for (auto field_json_obj : doc.get_object()) {
        auto key = std::string(static_cast<std::string_view>(field_json_obj.unescaped_key()));
        auto value = field_json_obj.value();
        ToLower(key);
        if (!possible_keys_set.erase(key)) {
            response["error_code"] = ErrorCode::kInvalidExpression;
            response["error_msg"] = fmt::format("Unknown or duplicate MatchDense expression key: {}", key);
            return nullptr;
        }
        if (key == "match_method") {
            std::string match_method = value.get<std::string>();
            ToLower(match_method);
            if (match_method != "dense") {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = fmt::format("MatchDense expression match_method should be dense, but got: {}", match_method);
                return nullptr;
            }
        } else if (key == "fields") {
            auto column_expr = std::make_unique<ColumnExpr>();
            auto column_str = value.get<std::string>();
            column_expr->names_.push_back(std::move(column_str));
            knn_expr->column_expr_ = column_expr.release();
        } else if (key == "query_vector") {
            query_vector_json = value.raw_json();
            has_query_vector = true;
        } else if (key == "fde") {
            fde_json = value.raw_json();
            has_fde = true;
        } else if (key == "element_type") {
            std::string element_type = value.get<std::string>();
            ToUpper(element_type);
            try {
                knn_expr->embedding_data_type_ = EmbeddingT::String2EmbeddingDataType(element_type);
            } catch (std::exception &e) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = fmt::format("Not supported vector element type: {}", element_type);
                return nullptr;
            }
        } else if (key == "metric_type") {
            std::string metric_type = value.get<std::string>();
            ToLower(metric_type);
            if (!knn_expr->InitDistanceType(metric_type.c_str())) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = fmt::format("Unknown knn distance: {}", metric_type);
                return nullptr;
            }
        } else if (key == "topn") {
            if (!value.is_integer()) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "MatchDense topn field should be integer";
                return nullptr;
            }
            topn = value.get<i64>();
        } else if (key == "params") {
            if (value.type() != simdjson::json_type::object) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "MatchDense params should be object";
                return nullptr;
            }
            for (auto param_it : value.get_object()) {
                auto param_k = std::string(static_cast<std::string_view>(param_it.unescaped_key()));
                auto param_v = param_it.value();
                if (param_k == "filter") {
                    if (knn_expr->filter_expr_) {
                        response["error_code"] = ErrorCode::kInvalidExpression;
                        response["error_msg"] = "More than one filter field in match dense params.";
                        return nullptr;
                    }
                    knn_expr->filter_expr_ = ParseFilter(param_v.raw_json(), http_status, response);
                    if (!knn_expr->filter_expr_) {
                        return nullptr;
                    }
                    // do not put it into opt_params_
                    continue;
                }
                if (param_k == "index_name") {
                    knn_expr->index_name_ = param_v.get<std::string>();
                    continue;
                }
                if (param_k == "ignore_index" && (std::string)param_v.get<std::string>() == "true") {
                    knn_expr->ignore_index_ = true;
                    continue;
                }
                if (knn_expr->opt_params_ == nullptr) {
                    knn_expr->opt_params_ = new std::vector<InitParameter *>();
                }
                auto parameter = std::make_unique<InitParameter>();
                parameter->param_name_ = param_k;
                parameter->param_value_ = param_v.get<std::string>();
                knn_expr->opt_params_->emplace_back(parameter.release());
            }
        }
    }
    // "params" is optional
    possible_keys_set.erase("params");
    // Either "query_vector" or "fde" is required, but not both
    possible_keys_set.erase("query_vector");
    possible_keys_set.erase("fde");

    // Validate query_vector vs fde usage
    if (has_query_vector && has_fde) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = "MatchDense expression cannot have both 'query_vector' and 'fde' fields";
        return nullptr;
    }
    if (!has_query_vector && !has_fde) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = "MatchDense expression must have either 'query_vector' or 'fde' field";
        return nullptr;
    }

    // check if all required fields are set
    if (!possible_keys_set.empty()) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] =
            fmt::format("Invalid MatchDense expression: following fields are required but not found: {}", fmt::join(possible_keys_set, ", "));
        return nullptr;
    }
    if (topn <= 0) {
        response["error_code"] = ErrorCode::kInvalidTopKType;
        response["error_msg"] = "MatchDense expression topn field should be positive integer";
        return nullptr;
    }
    knn_expr->topn_ = topn;

    if (has_query_vector) {
        // Traditional query_vector approach
        const auto [dimension, embedding_ptr] = ParseVector(query_vector_json, knn_expr->embedding_data_type_, http_status, response);
        if (embedding_ptr == nullptr) {
            return nullptr;
        }
        knn_expr->dimension_ = dimension;
        knn_expr->embedding_data_ptr_ = embedding_ptr;
    } else if (has_fde) {
        // FDE function approach
        auto fde_expr = ParseFDEFunction(fde_json, knn_expr->embedding_data_type_, http_status, response);
        if (!fde_expr) {
            return nullptr;
        }
        knn_expr->query_embedding_expr_ = std::move(fde_expr);
        knn_expr->embedding_data_type_str_ = EmbeddingT::EmbeddingDataType2String(knn_expr->embedding_data_type_);
        // Dimension will be determined at runtime by the FDE function
        knn_expr->dimension_ = -1; // Placeholder for runtime determination
    }

    return knn_expr;
}

std::unique_ptr<MatchExpr> HTTPSearch::ParseMatchText(std::string_view json_sv, HTTPStatus &http_status, nlohmann::json &response) {
    simdjson::padded_string json_pad(json_sv);
    simdjson::parser parser;
    simdjson::document doc = parser.iterate(json_pad);

    if (doc.type() != simdjson::json_type::object) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = "MatchText field should be object";
        return nullptr;
    }
    auto match_expr = std::make_unique<MatchExpr>();
    i64 topn = -1;
    std::string extra_params{};
    // must have: "match_method", "fields", "matching_text", "topn"
    // may have: "params"
    constexpr std::array possible_keys{"match_method", "fields", "matching_text", "topn", "params"};
    std::set<std::string> possible_keys_set(possible_keys.begin(), possible_keys.end());
    for (auto field_json_obj : doc.get_object()) {
        auto key = std::string(static_cast<std::string_view>(field_json_obj.unescaped_key()));
        auto value = field_json_obj.value();
        ToLower(key);
        if (!possible_keys_set.erase(key)) {
            response["error_code"] = ErrorCode::kInvalidExpression;
            response["error_msg"] = fmt::format("Unknown or duplicate MatchText expression key: {}", key);
            return nullptr;
        }
        if (key == "match_method") {
            std::string match_method = value.get<std::string>();
            ToLower(match_method);
            if (match_method != "text") {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = fmt::format("MatchText expression match_method should be text, but got: {}", match_method);
                return nullptr;
            }
        } else if (key == "fields") {
            match_expr->fields_ = value.get<std::string>();
        } else if (key == "matching_text") {
            match_expr->matching_text_ = value.get<std::string>();
        } else if (key == "topn") {
            if (!value.is_integer()) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "MatchText topn field should be integer";
                return nullptr;
            }
            topn = value.get<i64>();
        } else if (key == "params") {
            if (value.type() != simdjson::json_type::object) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "MatchText params should be object";
                return nullptr;
            }
            for (auto param : value.get_object()) {
                auto param_k = std::string(static_cast<std::string_view>(param.unescaped_key()));
                auto param_v = param.value();
                if (param_k == "index_names") {
                    match_expr->index_names_ = param_v.get<std::string>();
                    continue;
                }
                if (param_k == "filter") {
                    if (match_expr->filter_expr_) {
                        response["error_code"] = ErrorCode::kInvalidExpression;
                        response["error_msg"] = "More than one filter field in match text params.";
                        return nullptr;
                    }
                    match_expr->filter_expr_ = ParseFilter(param_v.raw_json(), http_status, response);
                    if (!match_expr->filter_expr_) {
                        return nullptr;
                    }
                    // do not put it into extra_params
                    continue;
                }
                extra_params += fmt::format(";{}={}", param_k, (std::string)param_v.get<std::string>());
            }
        }
    }
    // "params" is optional
    possible_keys_set.erase("params");
    // check if all required fields are set
    if (!possible_keys_set.empty()) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] =
            fmt::format("Invalid MatchText expression: following fields are required but not found: {}", fmt::join(possible_keys_set, ", "));
        return nullptr;
    }
    if (topn <= 0) {
        response["error_code"] = ErrorCode::kInvalidTopKType;
        response["error_msg"] = "MatchText expression topn field should be positive integer";
        return nullptr;
    }
    match_expr->options_text_ = fmt::format("topn={}{}", topn, extra_params);
    return match_expr;
}

std::unique_ptr<MatchTensorExpr> HTTPSearch::ParseMatchTensor(std::string_view json_sv, HTTPStatus &http_status, nlohmann::json &response) {
    simdjson::padded_string json_pad(json_sv);
    simdjson::parser parser;
    simdjson::document doc = parser.iterate(json_pad);

    if (doc.type() != simdjson::json_type::object) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = "MatchTensor field should be object";
        return nullptr;
    }
    auto match_tensor_expr = std::make_unique<MatchTensorExpr>();
    match_tensor_expr->SetSearchMethodStr("maxsim");
    std::string element_type{};
    std::shared_ptr<ConstantExpr> tensor_expr{};
    i64 topn = -1;
    std::string extra_params{};
    // must have: "match_method", "fields", "query_tensor", "element_type", "topn"
    // may have: "params"
    constexpr std::array possible_keys{"match_method", "field", "query_tensor", "element_type", "topn", "params"};
    std::set<std::string> possible_keys_set(possible_keys.begin(), possible_keys.end());
    for (auto field_json_obj : doc.get_object()) {
        auto key = std::string(static_cast<std::string_view>(field_json_obj.unescaped_key()));
        auto value = field_json_obj.value();
        ToLower(key);
        if (!possible_keys_set.erase(key)) {
            response["error_code"] = ErrorCode::kInvalidExpression;
            response["error_msg"] = fmt::format("Unknown or duplicate MatchTensor expression key: {}", key);
            return nullptr;
        }
        if (key == "match_method") {
            std::string match_method = value.get<std::string>();
            ToLower(match_method);
            if (match_method != "tensor") {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = fmt::format("MatchTensor expression match_method should be tensor, but got: {}", match_method);
                return nullptr;
            }
        } else if (key == "field") {
            auto column_str = value.get<std::string>();
            auto column_expr = std::make_unique<ColumnExpr>();
            column_expr->names_.push_back(std::move(column_str));
            match_tensor_expr->column_expr_ = std::move(column_expr);
        } else if (key == "query_tensor") {
            simdjson::padded_string find_pad(json_sv);
            simdjson::document find_doc = parser.iterate(find_pad);
            if (simdjson::value val; find_doc["element_type"].get(val) != simdjson::SUCCESS) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "Missing element_type for query_tensor";
                return nullptr;
            }
            try {
                tensor_expr = BuildConstantExprFromJson(value.raw_json());
            } catch (std::exception &e) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = fmt::format("Invalid query_tensor, error info: {}", e.what());
                return nullptr;
            }
        } else if (key == "element_type") {
            element_type = value.get<std::string>();
        } else if (key == "topn") {
            if (!value.is_integer()) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "MatchTensor topn field should be integer";
                return nullptr;
            }
            topn = value.get<i64>();
        } else if (key == "params") {
            if (value.type() != simdjson::json_type::object) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "MatchTensor params should be object";
                return nullptr;
            }
            for (auto param : value.get_object()) {
                auto param_k = std::string(static_cast<std::string_view>(param.unescaped_key()));
                auto param_v = param.value();
                if (param_k == "index_name") {
                    match_tensor_expr->index_name_ = param_v.get<std::string>();
                    continue;
                }
                if (param_k == "ignore_index" && (std::string)param_v.get<std::string>() == "true") {
                    match_tensor_expr->ignore_index_ = true;
                    continue;
                }
                if (param_k == "filter") {
                    if (match_tensor_expr->filter_expr_) {
                        response["error_code"] = ErrorCode::kInvalidExpression;
                        response["error_msg"] = "More than one filter field in match tensor params.";
                        return nullptr;
                    }
                    match_tensor_expr->filter_expr_ = ParseFilter(param_v.raw_json(), http_status, response);
                    if (!match_tensor_expr->filter_expr_) {
                        return nullptr;
                    }
                    // do not put it into extra_params
                    continue;
                }
                extra_params += fmt::format(";{}={}", param_k, static_cast<std::string>(param_v.get<std::string>()));
            }
        }
    }
    // "params" is optional
    possible_keys_set.erase("params");
    // check if all required fields are set
    if (!possible_keys_set.empty()) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] =
            fmt::format("Invalid MatchTensor expression: following fields are required but not found: {}", fmt::join(possible_keys_set, ", "));
        return nullptr;
    }
    if (topn <= 0) {
        response["error_code"] = ErrorCode::kInvalidTopKType;
        response["error_msg"] = "MatchTensor expression topn field should be positive integer";
        return nullptr;
    }
    match_tensor_expr->options_text_ = fmt::format("topn={}{}", topn, extra_params);
    try {
        match_tensor_expr->SetQueryTensorStr(std::move(element_type), tensor_expr.get());
    } catch (std::exception &e) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = fmt::format("Invalid query_tensor, error info: {}", e.what());
        return nullptr;
    }
    return match_tensor_expr;
}

std::unique_ptr<MatchSparseExpr> HTTPSearch::ParseMatchSparse(std::string_view json_sv, HTTPStatus &http_status, nlohmann::json &response) {
    simdjson::padded_string json_pad(json_sv);
    simdjson::parser parser;
    simdjson::document doc = parser.iterate(json_pad);

    if (doc.type() != simdjson::json_type::object) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = "MatchSparse field should be object";
        return nullptr;
    }
    auto match_sparse_expr = std::make_unique<MatchSparseExpr>();
    auto *opt_params_ptr = new std::vector<InitParameter *>();
    DeferFn release_opt([&]() {
        if (opt_params_ptr != nullptr) {
            for (auto &params_ptr : *opt_params_ptr) {
                delete params_ptr;
            }
            delete opt_params_ptr;
            opt_params_ptr = nullptr;
        }
    });
    i64 topn = -1;
    // must have: "match_method", "fields", "query_vector", "metric_type", "topn"
    // may have: "params"
    constexpr std::array possible_keys{"match_method", "fields", "query_vector", "metric_type", "topn", "params"};
    std::set<std::string> possible_keys_set(possible_keys.begin(), possible_keys.end());
    for (auto field_json_obj : doc.get_object()) {
        auto key = std::string(static_cast<std::string_view>(field_json_obj.unescaped_key()));
        auto value = field_json_obj.value();
        ToLower(key);
        if (!possible_keys_set.erase(key)) {
            response["error_code"] = ErrorCode::kInvalidExpression;
            response["error_msg"] = fmt::format("Unknown or duplicate MatchSparse expression key: {}", key);
            return nullptr;
        }
        if (key == "match_method") {
            std::string match_method = value.get<std::string>();
            ToLower(match_method);
            if (match_method != "sparse") {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = fmt::format("MatchSparse expression match_method should be sparse, but got: {}", match_method);
                return nullptr;
            }
        } else if (key == "fields") {
            auto column_expr = std::make_unique<ColumnExpr>();
            auto column_str = value.get<std::string>();
            column_expr->names_.push_back(std::move(column_str));
            match_sparse_expr->column_expr_ = std::move(column_expr);
        } else if (key == "query_vector") {
            std::unique_ptr<ConstantExpr> query_sparse_expr = ParseSparseVector(value.raw_json(), http_status, response);
            if (!query_sparse_expr) {
                return nullptr;
            }
            ConstantExpr *const_sparse_expr = query_sparse_expr.release();
            match_sparse_expr->SetQuerySparse(const_sparse_expr);
            assert(const_sparse_expr == nullptr);
        } else if (key == "metric_type") {
            try {
                match_sparse_expr->SetMetricType(value.get<std::string>());
            } catch (std::exception &e) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = fmt::format("Invalid metric_type: {}, error info: {}",
                                                    static_cast<std::string>(field_json_obj.value().get<std::string>()),
                                                    e.what());
                return nullptr;
            }
        } else if (key == "topn") {
            if (!value.is_integer()) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "MatchSparse topn field should be integer";
                return nullptr;
            }
            topn = value.get<i64>();
        } else if (key == "params") {
            if (value.type() != simdjson::json_type::object) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "MatchSparse params should be object";
                return nullptr;
            }
            for (auto param : value.get_object()) {
                auto param_k = std::string(static_cast<std::string_view>(param.unescaped_key()));
                auto param_v = param.value();
                if (param_k == "filter") {
                    if (match_sparse_expr->filter_expr_) {
                        response["error_code"] = ErrorCode::kInvalidExpression;
                        response["error_msg"] = "More than one filter field in match sparse params.";
                        return nullptr;
                    }
                    match_sparse_expr->filter_expr_ = ParseFilter(param_v.raw_json(), http_status, response);
                    if (!match_sparse_expr->filter_expr_) {
                        return nullptr;
                    }
                    // do not put it into opt_params_ptr
                    continue;
                }
                if (param_k == "index_name") {
                    match_sparse_expr->index_name_ = param_v.get<std::string>();
                    continue;
                }
                if (param_k == "ignore_index" && static_cast<std::string>(param_v.get<std::string>()) == "true") {
                    match_sparse_expr->ignore_index_ = true;
                    continue;
                }
                auto *init_parameter = new InitParameter();
                init_parameter->param_name_ = param_k;
                init_parameter->param_value_ = param_v.get<std::string>();
                opt_params_ptr->emplace_back(init_parameter);
            }
        }
    }
    // "params" is optional
    possible_keys_set.erase("params");
    // check if all required fields are set
    if (!possible_keys_set.empty()) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] =
            fmt::format("Invalid MatchSparse expression: following fields are required but not found: {}", fmt::join(possible_keys_set, ", "));
        return nullptr;
    }
    if (topn <= 0) {
        response["error_code"] = ErrorCode::kInvalidTopKType;
        response["error_msg"] = "MatchSparse expression topn field should be positive integer";
        return nullptr;
    }
    match_sparse_expr->SetOptParams(topn, opt_params_ptr);
    opt_params_ptr = nullptr;
    return match_sparse_expr;
}

std::unique_ptr<ConstantExpr> HTTPSearch::ParseSparseVector(std::string_view json_sv, HTTPStatus &http_status, nlohmann::json &response) {
    simdjson::padded_string json_pad(json_sv);
    simdjson::parser parser;
    simdjson::document doc = parser.iterate(json_pad);

    if (doc.type() != simdjson::json_type::object) {
        response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
        response["error_msg"] = "Sparse vector should be object";
        return nullptr;
    }
    if (doc.count_fields() == 0) {
        response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
        response["error_msg"] = "Empty sparse vector, cannot decide type";
        return nullptr;
    }

    std::unique_ptr<ConstantExpr> const_sparse_expr{};
    {
        simdjson::padded_string find_pad(json_sv);
        simdjson::document find_doc = parser.iterate(find_pad);
        simdjson::field first_kv = *find_doc.get_object().begin();
        if (first_kv.value().type() != simdjson::json_type::number) {
            response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
            response["error_msg"] = "Sparse value element type error";
            return nullptr;
        }
        switch (first_kv.value().get_number_type()) {
            case simdjson::number_type::unsigned_integer:
            case simdjson::number_type::signed_integer: {
                const_sparse_expr = std::make_unique<ConstantExpr>(LiteralType::kLongSparseArray);
                break;
            }
            case simdjson::number_type::floating_point_number: {
                const_sparse_expr = std::make_unique<ConstantExpr>(LiteralType::kDoubleSparseArray);
                break;
            }
            default: {
                response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                response["error_msg"] = "Sparse value element type error";
                return nullptr;
            }
        }
    }

    for (std::unordered_set<i64> key_set; auto sparse_it : doc.get_object()) {
        auto sparse_k = std::string(static_cast<std::string_view>(sparse_it.unescaped_key()));
        auto sparse_v = sparse_it.value();
        i64 key_val = {};
        try {
            key_val = std::stoll(sparse_k);
        } catch (std::exception &e) {
            response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
            response["error_msg"] = fmt::format("Error when try to cast sparse key '{}' to integer, error info: {}", sparse_k, e.what());
            return nullptr;
        }
        if (const auto [_, insert_ok] = key_set.insert(key_val); !insert_ok) {
            response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
            response["error_msg"] = fmt::format("Duplicate key {} in sparse array!", key_val);
            return nullptr;
        }
        bool good_v = false;
        switch (sparse_v.get_number_type()) {
            case simdjson::number_type::unsigned_integer:
            case simdjson::number_type::signed_integer: {
                if (const_sparse_expr->literal_type_ == LiteralType::kLongSparseArray) {
                    const_sparse_expr->long_sparse_array_.first.push_back(key_val);
                    const_sparse_expr->long_sparse_array_.second.push_back(sparse_v.get<i64>());
                    good_v = true;
                }
                break;
            }
            case simdjson::number_type::floating_point_number: {
                if (const_sparse_expr->literal_type_ == LiteralType::kDoubleSparseArray) {
                    const_sparse_expr->double_sparse_array_.first.push_back(key_val);
                    const_sparse_expr->double_sparse_array_.second.push_back(sparse_v.get<double>());
                    good_v = true;
                }
                break;
            }
            default: {
                break;
            }
        }
        if (!good_v) {
            response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
            response["error_msg"] = "Sparse value element type error";
            return nullptr;
        }
    }
    return const_sparse_expr;
}

std::tuple<i64, void *>
HTTPSearch::ParseVector(std::string_view json_sv, EmbeddingDataType elem_type, HTTPStatus &http_status, nlohmann::json &response) {
    simdjson::padded_string json_pad(json_sv);
    simdjson::parser parser;
    simdjson::document doc = parser.iterate(json_pad);

    if (doc.type() != simdjson::json_type::array) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = "Can't recognize embedding/vector.";
        return {0, nullptr};
    }
    size_t dimension = doc.count_elements();
    if (dimension == 0) {
        response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
        response["error_msg"] = "Empty embedding data";
        return {0, nullptr};
    }

    switch (elem_type) {
        case EmbeddingDataType::kElemBit: {
            if (dimension % 8 != 0) {
                response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                response["error_msg"] = "bit embeddings should have dimension of times of 8";
                return {0, nullptr};
            }
            u8 *embedding_data_ptr = new u8[dimension / 8];
            DeferFn defer_free_embedding([&]() {
                if (embedding_data_ptr != nullptr) {
                    delete[] embedding_data_ptr;
                    embedding_data_ptr = nullptr;
                }
            });
            for (size_t i = 0; i < dimension / 8; ++i) {
                u8 unit = 0;
                for (size_t bit_idx = 0; bit_idx < 8; bit_idx++) {
                    simdjson::padded_string sub_pad(json_sv);
                    simdjson::parser sub_parser;
                    simdjson::document sub_doc = sub_parser.iterate(sub_pad);
                    auto value_ref = sub_doc.at(i * 8 + bit_idx);
                    if (value_ref.type() != simdjson::json_type::number) {
                        response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                        response["error_msg"] = "Embedding element type should be integer";
                        return {0, nullptr};
                    }
                    switch (value_ref.get_number_type()) {
                        case simdjson::number_type::signed_integer:
                        case simdjson::number_type::unsigned_integer: {
                            if (value_ref.get<i64>() > 0) {
                                unit |= (1 << bit_idx);
                            }
                            break;
                        }
                        default: {
                            response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                            response["error_msg"] = "Embedding element type should be integer";
                            return {0, nullptr};
                        }
                    }
                }
                embedding_data_ptr[i] = unit;
            }
            u8 *res = embedding_data_ptr;
            embedding_data_ptr = nullptr;
            return {dimension, res};
        }
        case EmbeddingDataType::kElemInt32: {
            i32 *embedding_data_ptr = new i32[dimension];
            DeferFn defer_free_embedding([&]() {
                if (embedding_data_ptr != nullptr) {
                    delete[] embedding_data_ptr;
                    embedding_data_ptr = nullptr;
                }
            });
            for (size_t idx = 0; auto value_ref : doc.get_array()) {
                if (value_ref.type() != simdjson::json_type::number) {
                    response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                    response["error_msg"] = "Embedding element type should be integer";
                    return {0, nullptr};
                }
                switch (value_ref.get_number_type()) {
                    case simdjson::number_type::signed_integer:
                    case simdjson::number_type::unsigned_integer: {
                        embedding_data_ptr[idx++] = value_ref.get<i64>();
                        break;
                    }
                    default: {
                        response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                        response["error_msg"] = "Embedding element type should be integer";
                        return {0, nullptr};
                    }
                }
            }

            i32 *res = embedding_data_ptr;
            embedding_data_ptr = nullptr;
            return {dimension, res};
        }
        case EmbeddingDataType::kElemInt8: {
            i8 *embedding_data_ptr = new i8[dimension];
            DeferFn defer_free_embedding([&]() {
                if (embedding_data_ptr != nullptr) {
                    delete[] embedding_data_ptr;
                    embedding_data_ptr = nullptr;
                }
            });
            for (size_t idx = 0; auto value_ref : doc.get_array()) {
                if (value_ref.type() != simdjson::json_type::number) {
                    response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                    response["error_msg"] = "Embedding element type should be integer";
                    return {0, nullptr};
                }
                switch (value_ref.get_number_type()) {
                    case simdjson::number_type::signed_integer:
                    case simdjson::number_type::unsigned_integer: {
                        embedding_data_ptr[idx++] = value_ref.get<i64>();
                        break;
                    }
                    default: {
                        response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                        response["error_msg"] = "Embedding element type should be integer";
                        return {0, nullptr};
                    }
                }
            }

            i8 *res = embedding_data_ptr;
            embedding_data_ptr = nullptr;
            return {dimension, res};
        }
        case EmbeddingDataType::kElemUInt8: {
            u8 *embedding_data_ptr = new u8[dimension];
            DeferFn defer_free_embedding([&]() {
                if (embedding_data_ptr != nullptr) {
                    delete[] embedding_data_ptr;
                    embedding_data_ptr = nullptr;
                }
            });
            for (size_t idx = 0; auto value_ref : doc.get_array()) {
                if (value_ref.type() != simdjson::json_type::number) {
                    response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                    response["error_msg"] = "Embedding element type should be integer";
                    return {0, nullptr};
                }
                switch (value_ref.get_number_type()) {
                    case simdjson::number_type::signed_integer:
                    case simdjson::number_type::unsigned_integer: {
                        embedding_data_ptr[idx++] = value_ref.get<i64>();
                        break;
                    }
                    default: {
                        response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                        response["error_msg"] = "Embedding element type should be integer";
                        return {0, nullptr};
                    }
                }
            }

            u8 *res = embedding_data_ptr;
            embedding_data_ptr = nullptr;
            return {dimension, res};
        }
        case EmbeddingDataType::kElemFloat: {
            f32 *embedding_data_ptr = new f32[dimension];
            DeferFn defer_free_embedding([&]() {
                if (embedding_data_ptr != nullptr) {
                    delete[] embedding_data_ptr;
                    embedding_data_ptr = nullptr;
                }
            });
            for (size_t idx = 0; auto value_ref : doc.get_array()) {
                if (value_ref.type() != simdjson::json_type::number) {
                    response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                    response["error_msg"] = "Embedding element type should be float";
                    return {0, nullptr};
                }
                switch (value_ref.get_number_type()) {
                    case simdjson::number_type::floating_point_number: {
                        embedding_data_ptr[idx++] = value_ref.get<double>();
                        break;
                    }
                    case simdjson::number_type::signed_integer:
                    case simdjson::number_type::unsigned_integer: {
                        embedding_data_ptr[idx++] = value_ref.get<i64>();
                        break;
                    }
                    default: {
                        response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                        response["error_msg"] = "Embedding element type should be float";
                        return {0, nullptr};
                    }
                }
            }

            f32 *res = embedding_data_ptr;
            embedding_data_ptr = nullptr;
            return {dimension, res};
        }
        case EmbeddingDataType::kElemFloat16: {
            Float16T *embedding_data_ptr = new Float16T[dimension];
            DeferFn defer_free_embedding([&]() {
                if (embedding_data_ptr != nullptr) {
                    delete[] embedding_data_ptr;
                    embedding_data_ptr = nullptr;
                }
            });
            for (size_t idx = 0; auto value_ref : doc.get_array()) {
                if (value_ref.type() != simdjson::json_type::number) {
                    response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                    response["error_msg"] = "Embedding element type should be float";
                    return {0, nullptr};
                }
                switch (value_ref.get_number_type()) {
                    case simdjson::number_type::floating_point_number: {
                        embedding_data_ptr[idx++] = value_ref.get<double>();
                        break;
                    }
                    case simdjson::number_type::signed_integer:
                    case simdjson::number_type::unsigned_integer: {
                        embedding_data_ptr[idx++] = value_ref.get<i64>();
                        break;
                    }
                    default: {
                        response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                        response["error_msg"] = "Embedding element type should be float";
                        return {0, nullptr};
                    }
                }
            }

            Float16T *res = embedding_data_ptr;
            embedding_data_ptr = nullptr;
            return {dimension, res};
        }
        case EmbeddingDataType::kElemBFloat16: {
            BFloat16T *embedding_data_ptr = new BFloat16T[dimension];
            DeferFn defer_free_embedding([&]() {
                if (embedding_data_ptr != nullptr) {
                    delete[] embedding_data_ptr;
                    embedding_data_ptr = nullptr;
                }
            });
            for (size_t idx = 0; auto value_ref : doc.get_array()) {
                if (value_ref.type() != simdjson::json_type::number) {
                    response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                    response["error_msg"] = "Embedding element type should be float";
                    return {0, nullptr};
                }
                switch (value_ref.get_number_type()) {
                    case simdjson::number_type::floating_point_number: {
                        embedding_data_ptr[idx++] = value_ref.get<double>();
                        break;
                    }
                    case simdjson::number_type::signed_integer:
                    case simdjson::number_type::unsigned_integer: {
                        embedding_data_ptr[idx++] = value_ref.get<i64>();
                        break;
                    }
                    default: {
                        response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
                        response["error_msg"] = "Embedding element type should be float";
                        return {0, nullptr};
                    }
                }
            }

            BFloat16T *res = embedding_data_ptr;
            embedding_data_ptr = nullptr;
            return {dimension, res};
        }
        default: {
            response["error_code"] = ErrorCode::kInvalidEmbeddingDataType;
            response["error_msg"] = "Only support float as the embedding data type";
            return {0, nullptr};
        }
    }
}

std::unique_ptr<FunctionExpr>
HTTPSearch::ParseFDEFunction(std::string_view json_sv, EmbeddingDataType elem_type, HTTPStatus &http_status, nlohmann::json &response) {
    simdjson::padded_string json_pad(json_sv);
    simdjson::parser parser;
    simdjson::document doc = parser.iterate(json_pad);

    if (doc.type() != simdjson::json_type::object) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = "FDE field should be object";
        return nullptr;
    }

    std::string_view query_tensor_json;
    i64 target_dimension = -1;

    // must have: "query_tensor", "target_dimension"
    constexpr std::array possible_keys{"query_tensor", "target_dimension"};
    std::set<std::string> possible_keys_set(possible_keys.begin(), possible_keys.end());

    for (auto field_json_obj : doc.get_object()) {
        std::string key = std::string((std::string_view)field_json_obj.unescaped_key());
        auto value = field_json_obj.value();
        ToLower(key);
        if (!possible_keys_set.erase(key)) {
            response["error_code"] = ErrorCode::kInvalidExpression;
            response["error_msg"] = fmt::format("Unknown or duplicate FDE field: {}", key);
            return nullptr;
        }
        if (key == "query_tensor") {
            query_tensor_json = value.raw_json();
        } else if (key == "target_dimension") {
            if (!value.is_integer()) {
                response["error_code"] = ErrorCode::kInvalidExpression;
                response["error_msg"] = "FDE target_dimension field should be integer";
                return nullptr;
            }
            target_dimension = value.get<i64>();
        }
    }

    // check if all required fields are set
    if (!possible_keys_set.empty()) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] =
            fmt::format("Invalid FDE expression: following fields are required but not found: {}", fmt::join(possible_keys_set, ", "));
        return nullptr;
    }

    if (target_dimension <= 0) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = "FDE target_dimension should be positive integer";
        return nullptr;
    }

    // Parse the query tensor
    std::shared_ptr<ConstantExpr> tensor_expr;
    try {
        tensor_expr = BuildConstantExprFromJson(query_tensor_json);
        if (!tensor_expr) {
            response["error_code"] = ErrorCode::kInvalidExpression;
            response["error_msg"] = "Failed to parse query_tensor";
            return nullptr;
        }
    } catch (std::exception &e) {
        response["error_code"] = ErrorCode::kInvalidExpression;
        response["error_msg"] = fmt::format("Invalid query_tensor, error info: {}", e.what());
        return nullptr;
    }

    // Create FDE function expression
    auto fde_function = std::make_unique<FunctionExpr>();
    fde_function->func_name_ = "FDE";
    fde_function->arguments_ = new std::vector<ParsedExpr *>();

    // Transfer ownership of tensor_expr to FunctionExpr
    // Create a new ConstantExpr by moving from the shared_ptr content
    auto tensor_arg = new ConstantExpr(std::move(*tensor_expr));
    fde_function->arguments_->push_back(tensor_arg);

    // Add target dimension as second argument
    auto target_dim_expr = std::make_unique<ConstantExpr>(LiteralType::kInteger);
    target_dim_expr->integer_value_ = target_dimension;
    fde_function->arguments_->push_back(target_dim_expr.release());

    return fde_function;
}

} // namespace infinity
